package com.mlog.njyjy.ndop.share.kafka.oauth;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.io.*;
import java.lang.reflect.Field;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 * .
 * </p>
 *
 * @author <a href="mailto:yingxiufeng@mlogcn.com">alex</a>
 * @version v1.0, 2020/7/16
 */
public class OauthRequest {

    private static final Logger log = LoggerFactory.getLogger(OauthRequest.class);

    private static final String OAUTH_PROPERTIES_PATH = "OAUTH_PROPERTIES_PATH";

    private static final String OAUTH_PROPERTIES = "kafka-oauth.properties";

    private KafkaOauthConfig oauthConfig = new KafkaOauthConfig();

    private static Time time = Time.SYSTEM;


    private OauthRequest() {
    }

    /**
     * 创建一个 对象
     *
     * @return
     */
    public static OauthRequest create() {
        OauthRequest request = new OauthRequest();
        request.loadConfig();
        return request;
    }

    /**
     * TODO
     *
     * @param clientId
     * @return
     */
    public OauthBearerTokenJwt login(String clientId) {
        OauthBearerTokenJwt result = null;
        try {
            acceptUnsecureServer();
            long callTime = new SystemTime().milliseconds();

            //Mount POST data
            String grantType = "grant_type=" + oauthConfig.getGrantType();
            String scope = "scope=" + oauthConfig.getScope();
            String postDataStr = grantType + "&" + scope;

            log.info("Try to login with oauth!");
            Map<String, Object> resp = doHttpCall(oauthConfig.getTokenUrl(), postDataStr,
                    Base64.getEncoder().encodeToString((oauthConfig.getUsername() + ":" + oauthConfig.getPassword()).getBytes()));
            if (resp != null) {
                String accessToken = (String) resp.get("access_token");
                long expiresIn = ((Integer) resp.get("expires_in")).longValue();
                result = new OauthBearerTokenJwt(accessToken, expiresIn, callTime, clientId);
            } else {
                throw new Exception("with resp null at login");
            }
        } catch (Exception e) {
            log.error(e.getLocalizedMessage());
        }
        return result;
    }

    /**
     * 验证 token
     * TODO
     *
     * @param accessToken
     * @return
     */
    public OauthBearerTokenJwt introspectToken(String accessToken) {
        OauthBearerTokenJwt result = null;
        try {
            //Mount POST data
            String token = "token=" + accessToken;
            log.info("Try to introspect with oauth!");
            Map<String, Object> resp = null;
//            doHttpCall(oauthConfig.getIntrospectUrl(), token, );
//            if(OAUTH_WITH_SSL){
//                resp = doHttpsCall(OAUTH_INTROSPECT_SERVER + OAUTH_INTROSPECT_ENDPOINT, token, OAUTH_INTROSPECT_AUTHORIZATION);
//            }else{
//                resp = doHttpCall(OAUTH_INTROSPECT_SERVER + OAUTH_INTROSPECT_ENDPOINT, token, OAUTH_INTROSPECT_AUTHORIZATION);
//            }
            if (resp != null) {
                if ((boolean) resp.get("active")) {
                    result = new OauthBearerTokenJwt(resp, accessToken);
                } else {
                    throw new Exception("Expired Token");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }

    private Map<String, Object> doHttpCall(String urlStr, String postParameters, String oauthToken) {
        try {
            System.out.println("doHttpCall ->");
            acceptUnsecureServer();

            byte[] postData = postParameters.getBytes(StandardCharsets.UTF_8);
            int postDataLength = postData.length;

            URL url = new URL("http://" + urlStr);
            HttpURLConnection con = (HttpURLConnection) url.openConnection();
            con.setInstanceFollowRedirects(true);
            con.setRequestMethod("POST");
            con.setRequestProperty("Authorization", oauthToken);
            con.setRequestProperty("Content-Type", "application/x-www-form-urlencoded");
            con.setRequestProperty("charset", "utf-8");
            con.setRequestProperty("Content-Length", Integer.toString(postDataLength));
            con.setUseCaches(false);
            con.setDoOutput(true);

            try (DataOutputStream wr = new DataOutputStream(con.getOutputStream())) {
                wr.write(postData);
            }

            int responseCode = con.getResponseCode();
            if (responseCode == 200) {
                return handleJsonResponse(con.getInputStream());
            } else {
                throw new Exception("Return code " + responseCode);
            }
        } catch (Exception e) {
            log.error("at doHttpCall", e);
        }
        return null;
    }

    private void acceptUnsecureServer() {
        TrustManager[] trustAllCerts = new TrustManager[]{
                new X509TrustManager() {
                    @Override
                    public java.security.cert.X509Certificate[] getAcceptedIssuers() {
                        return null;
                    }

                    @Override
                    public void checkClientTrusted(
                            java.security.cert.X509Certificate[] certs, String authType) {
                    }

                    @Override
                    public void checkServerTrusted(
                            java.security.cert.X509Certificate[] certs, String authType) {
                    }
                }
        };
        try {
            SSLContext sc = SSLContext.getInstance("SSL");
            sc.init(null, trustAllCerts, new java.security.SecureRandom());
            HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory());
        } catch (NoSuchAlgorithmException e) {
            log.error("at acceptUnsecureServer :", e);
        } catch (KeyManagementException e) {
            log.error("at acceptUnsecureServer :", e);
        }
    }

    private Map<String, Object> handleJsonResponse(InputStream inputStream) {
        Map<String, Object> result = null;
        try {
            BufferedReader in = new BufferedReader(new InputStreamReader(inputStream));
            String inputLine;
            StringBuffer response = new StringBuffer();

            while ((inputLine = in.readLine()) != null) {
                response.append(inputLine);
            }
            in.close();

            String jsonResponse = response.toString();
            ObjectMapper objectMapper = new ObjectMapper();
            result = objectMapper.readValue(jsonResponse, new TypeReference<Map<String, Object>>() {
            });

        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }

    /**
     * load & parse
     */
    private void loadConfig() {
        // load properties
        try {
            Properties properties = new Properties();
            String propertiesPath = System.getenv(OAUTH_PROPERTIES_PATH);
            // 从环境变量加载 > 配置目录加载
            if (propertiesPath != null) {
                File file = new File(propertiesPath.concat(File.separator + OAUTH_PROPERTIES));
                if (file.exists()) {
                    properties.load(new FileInputStream(file));
                }
            } else {
                InputStream inputStream = Thread.currentThread().getContextClassLoader().getResourceAsStream(OAUTH_PROPERTIES);
                properties.load(inputStream);
            }
            if (!properties.isEmpty()) {
                // convert to pojo
                properties.keySet().stream().forEach(key -> {
                    String propName = String.valueOf(key);
                    String fieldName;
                    if (propName.contains(".")) {
                        // to fieldName
                        String[] keys = propName.split("\\.");
                        fieldName = keys[0] + Arrays.stream(keys).skip(1).map(EnvUtil::upperFirst).collect(Collectors.joining(""));
                    } else {
                        fieldName = propName;
                    }
                    try {
                        String propValue = properties.getProperty(propName);
                        Field field = KafkaOauthConfig.class.getDeclaredField(fieldName);
                        field.setAccessible(true);
                        if (List.class.equals(field.getType())) {
                            field.set(oauthConfig, Arrays.asList(propValue.split(",")));
                        } else {
                            field.set(oauthConfig, propValue);
                        }
                    } catch (NoSuchFieldException | IllegalAccessException e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (IOException e) {
            log.error("加载配置异常:{}", e.getLocalizedMessage());
        }
    }

    public KafkaOauthConfig getOauthConfig() {
        return this.oauthConfig;
    }

}
